package com.alison.datastream.chapter6_timeAndWaterMark.watermark;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MySourceForLaten implements SourceFunction<String> {

    private boolean isRunning = true;
    long curTime = 1553503181000L;

    long step = 1000L;

    boolean delayFlag = false;

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        while (isRunning) {
            if (curTime == 1553503191000L && !delayFlag) {
                System.out.println("source: h1,1553503187000");
                // 将数据源收集写入SourceContext
                sourceContext.collect("h1,1553503187000");
                Thread.sleep(500);
                delayFlag = true;
                continue;
            }
            if (curTime == 1553503191000L && delayFlag) {
                System.out.println("source: h1,1553503184000");
                // 将数据源收集写入SourceContext
                sourceContext.collect("h1,1553503184000");
                Thread.sleep(500);
                delayFlag = true;
                String data = "h1," + curTime;
                System.out.println("source: " + data);
                // 将数据源收集写入SourceContext
                sourceContext.collect(data);
                curTime += step;
                continue;
            }
            String data = "h1," + curTime;
            System.out.println("source: " + data);
            // 将数据源收集写入SourceContext
            sourceContext.collect(data);
            curTime += step;
            Thread.sleep(500);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}